package com.baomidou.springboot.spark;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.springboot.constant.Constants;
import com.baomidou.springboot.utils.DateUtils;
import com.baomidou.springboot.utils.NumberUtils;
import com.baomidou.springboot.utils.ParamUtils;
import com.baomidou.springboot.utils.SparkUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;

import java.util.*;

/**
 * Created by pcmmm on 17/6/15.
 */
public class PageViewRatio {

    /**
     * 页面切片生成与匹配算法
     * @param sc
     * @param sessionid2actionsRDD
     * @param taskParam
     * @return
     */
    public static JavaPairRDD<String, Integer> generateAndMatchPageSplit(
            JavaSparkContext sc,
            JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD,
            JSONObject taskParam) {
        String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
        final Broadcast<String> targetPageFlowBroadcast = sc.broadcast(targetPageFlow);

        return sessionid2actionsRDD.flatMapToPair(

                new PairFlatMapFunction<Tuple2<String,Iterable<Row>>, String, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Tuple2<String, Integer>> call(
                            Tuple2<String, Iterable<Row>> tuple)
                            throws Exception {
                        // 定义返回list
                        List<Tuple2<String, Integer>> list =
                                new ArrayList<Tuple2<String, Integer>>();
                        // 获取到当前session的访问行为的迭代器
                        Iterator<Row> iterator = tuple._2.iterator();
                        // 获取使用者指定的页面流
                        // 使用者指定的页面流，1,2,3,4,5,6,7
                        // 1->2的转化率是多少？2->3的转化率是多少？
                        String[] targetPages = targetPageFlowBroadcast.value().split(",");

                        // 这里，我们拿到的session的访问行为，默认情况下是乱序的
                        // 比如说，正常情况下，我们希望拿到的数据，是按照时间顺序排序的
                        // 但是问题是，默认是不排序的
                        // 所以，我们第一件事情，对session的访问行为数据按照时间进行排序

                        // 举例，反例
                        // 比如，3->5->4->10->7
                        // 3->4->5->7->10
                        // 排序

                        List<Row> rows = new ArrayList<Row>();
                        while(iterator.hasNext()) {
                            rows.add(iterator.next());
                        }

                        Collections.sort(rows, new Comparator<Row>() {

                            @Override
                            public int compare(Row o1, Row o2) {
                                String actionTime1 = o1.getString(4);
                                String actionTime2 = o2.getString(4);

                                Date date1 = DateUtils.parseTime(actionTime1);
                                Date date2 = DateUtils.parseTime(actionTime2);

                                return (int)(date1.getTime() - date2.getTime());
                            }

                        });

                        // 页面切片的生成，以及页面流的匹配
                        Long lastPageId = null;

                        for(Row row : rows) {
                            long pageid = row.getLong(3);

                            if(lastPageId == null) {
                                lastPageId = pageid;
                                continue;
                            }

                            // 生成一个页面切片
                            // 3,5,2,1,8,9
                            // lastPageId=3
                            // 5，切片，3_5

                            String pageSplit = lastPageId + "_" + pageid;

                            // 对这个切片判断一下，是否在用户指定的页面流中
                            for(int i = 1; i < targetPages.length; i++) {
                                // 比如说，用户指定的页面流是3,2,5,8,1
                                // 遍历的时候，从索引1开始，就是从第二个页面开始
                                // 3_2
                                String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];

                                if(pageSplit.equals(targetPageSplit)) {
                                    list.add(new Tuple2<String, Integer>(pageSplit, 1));
                                    break;
                                }
                            }

                            lastPageId = pageid;
                        }

                        return list;
                    }

                });
    }

    /**
     * 获取页面流中初始页面的pv
     * @param taskParam
     * @param sessionid2actionsRDD
     * @return
     */
    public static long getStartPagePv(JSONObject taskParam,
                                       JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD) {
        String targetPageFlow = ParamUtils.getParam(taskParam,
                Constants.PARAM_TARGET_PAGE_FLOW);
        final long startPageId = Long.valueOf(targetPageFlow.split(",")[0]);

        JavaRDD<Long> startPageRDD = sessionid2actionsRDD.flatMap(

                new FlatMapFunction<Tuple2<String,Iterable<Row>>, Long>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Long> call(
                            Tuple2<String, Iterable<Row>> tuple)
                            throws Exception {
                        List<Long> list = new ArrayList<Long>();

                        Iterator<Row> iterator = tuple._2.iterator();

                        while(iterator.hasNext()) {
                            Row row = iterator.next();
                            long pageid = row.getLong(3);

                            if(pageid == startPageId) {
                                list.add(pageid);
                            }
                        }

                        return list;
                    }

                });

        return startPageRDD.count();
    }

    /**
     * 计算页面切片转化率
     * @param pageSplitPvMap 页面切片pv
     * @param startPagePv 起始页面pv
     * @return
     */
    public static Map<String, Double> computePageSplitConvertRate(
            JSONObject taskParam,
            Map<String, Object> pageSplitPvMap,
            long startPagePv) {
        Map<String, Double> convertRateMap = new HashMap<String, Double>();

        String[] targetPages = ParamUtils.getParam(taskParam,
                Constants.PARAM_TARGET_PAGE_FLOW).split(",");

        long lastPageSplitPv = 0L;

        // 3,5,2,4,6
        // 3_5
        // 3_5 pv / 3 pv
        // 5_2 rate = 5_2 pv / 3_5 pv

        // 通过for循环，获取目标页面流中的各个页面切片（pv）
        for(int i = 1; i < targetPages.length; i++) {
            String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];
            long targetPageSplitPv = Long.valueOf(String.valueOf(
                    pageSplitPvMap.get(targetPageSplit)));

            double convertRate = 0.0;

            if(i == 1) {
                convertRate = NumberUtils.formatDouble(
                        (double)targetPageSplitPv / (double)startPagePv, 2);
            } else {
                convertRate = NumberUtils.formatDouble(
                        (double)targetPageSplitPv / (double)lastPageSplitPv, 2);
            }

            convertRateMap.put(targetPageSplit, convertRate);

            lastPageSplitPv = targetPageSplitPv;
        }

        return convertRateMap;
    }

    /**
     * 持久化转化率
     * @param convertRateMap
     */
    public static void persistConvertRate(long taskid,
                                           Map<String, Double> convertRateMap) {
        StringBuffer buffer = new StringBuffer("");

        for(Map.Entry<String, Double> convertRateEntry : convertRateMap.entrySet()) {
            String pageSplit = convertRateEntry.getKey();
            double convertRate = convertRateEntry.getValue();
            buffer.append(pageSplit + "=" + convertRate + "|");
        }

        String convertRate = buffer.toString();
        convertRate = convertRate.substring(0, convertRate.length() - 1);
        System.out.println(convertRate);
        /*PageSplitConvertRate pageSplitConvertRate = new PageSplitConvertRate();
        pageSplitConvertRate.setTaskid(taskid);
        pageSplitConvertRate.setConvertRate(convertRate);

        IPageSplitConvertRateDAO pageSplitConvertRateDAO = DAOFactory.getPageSplitConvertRateDAO();
        pageSplitConvertRateDAO.insert(pageSplitConvertRate);*/

    }
}

